package com.shujia.mr.wc1;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/*
    默认情况下，map任务读取数据的方式是一行一行读取的，map输入的键和值的类型是固定的
    键是LongWritable,值是Text

    输入：
        "hello flink spark hive hadoop"

    输出：
        <"hello",1L>
        <"flink",1L>
        <"spark",1L>
        <"hive",1L>
        <"hadoop",1L>

    编写map类的步骤：
        1）继承org.apache.hadoop.mapreduce包下的Mapper类，使用新版本的api
        2）定义好输入和输出的键值类型
        3）重写方法，
            必须要重写的方法：map，要在这个方法中定义map任务的处理逻辑
            可选的重写方法：setup，cleanup

    注意：
        1、map方法每一行数据都会执行一次
        2、setup方法只会在每一个map任务执行之前，执行一次

 */
public class WordCountMapper extends Mapper<LongWritable, Text,Text,LongWritable> {
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
       // 这个方法主要是针对每一个map任务执行之前要做的事情，例如建立数据库的连接
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //每一行的处理逻辑
        //context是hadoop上下文环境对象，该对象是hadoop自己提供，使用这个对象来获取hadoop的环境配置，以及将map数据发送到下一个阶段【环形缓冲区】

        //<0, "hello flink spark hive hadoop">
        //将Text类型转成java中的String类型
        String line = value.toString();
        String[] words = line.split(" "); // ["hello","flink","spark","hive","hadoop"]
        for (String word : words) {
            context.write(new Text(word),new LongWritable(1L)); //写入到环形缓冲区
        }
        //后续在环形缓冲区中的分区编号，快速排序，溢写，都不需要我们来实现，hadoop集群帮我们去实现
        //开发者只需要关心map任务的处理逻辑就可以了。

    }

    @Override
    protected void cleanup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        // 这个方法主要是针对每一个map任务结束之后要做的事情，例如关闭setup中创建的数据库连接
    }
}
